package com.qying.rpc.codec;

import com.qying.rpc.common.MiniRpcRequest;
import com.qying.rpc.common.MiniRpcResponse;
import com.qying.rpc.protocol.MiniRpcProtocol;
import com.qying.rpc.protocol.MsgHeader;
import com.qying.rpc.protocol.MsgType;
import com.qying.rpc.protocol.ProtocolConstants;
import com.qying.rpc.serialzation.RpcSerialzation;
import com.qying.rpc.serialzation.SerialzationFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;
import java.util.Objects;

public class MiniRpcDecoder extends ByteToMessageDecoder {

    /*
    +---------------------------------------------------------------+
    | 魔数 2byte | 协议版本号 1byte | 序列化算法 1byte | 报文类型 1byte  |
    +---------------------------------------------------------------+
    | 状态 1byte |        消息 ID 8byte     |      数据长度 4byte     |
    +---------------------------------------------------------------+
    |                   数据内容 （长度不定）                          |
    +---------------------------------------------------------------+
    */

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf msg, List<Object> list) throws Exception {
        if (msg.readableBytes() < ProtocolConstants.HEADER_TOTAL_LEN) {
            return;
        }

        msg.markReaderIndex();

        short magic = msg.readShort();

        if (magic != ProtocolConstants.MAGIC) {
            throw new IllegalArgumentException("magic number is error");
        }

        byte version = msg.readByte();
        byte serializeType = msg.readByte();
        byte msgType = msg.readByte();
        byte status = msg.readByte();
        long requestId = msg.readLong();

        int dataLength = msg.readInt();

        if (msg.readableBytes() < dataLength) {
            msg.resetReaderIndex();
            return;
        }

        byte[] data = new byte[dataLength];
        msg.readBytes(data);

        MsgType byType = MsgType.findByType(msgType);

        if (byType == null) {
            return;
        }

        MsgHeader header = new MsgHeader();
        header.setMagic(magic);
        header.setVersion(version);
        header.setSerialzation(serializeType);
        header.setStatus(status);
        header.setRequestId(requestId);
        header.setMsgType(msgType);
        header.setMsgLen(dataLength);

        RpcSerialzation rpcSerialZation = SerialzationFactory.getRpcSerialZation(serializeType);

        switch (byType) {
            case REQUEST:
                MiniRpcRequest request = rpcSerialZation.deserialize(data, MiniRpcRequest.class);
                if (Objects.nonNull(request)) {
                    MiniRpcProtocol<MiniRpcRequest> protocol = new MiniRpcProtocol<>();
                    protocol.setHeader(header);
                    protocol.setBody(request);
                    list.add(protocol);
                }
                break;
            case RESPONSE:
                MiniRpcResponse response = rpcSerialZation.deserialize(data, MiniRpcResponse.class);
                if (Objects.nonNull(response)) {
                    MiniRpcProtocol<MiniRpcResponse> protocol = new MiniRpcProtocol<>();
                    protocol.setHeader(header);
                    protocol.setBody(response);
                    list.add(protocol);
                }
                break;
            case HEAETBEAT:
                break;
        }

    }
}
